package org.zkdemo.actors;

import org.apache.log4j.Logger;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.context.ApplicationContext;
import org.zkdemo.messages.Message;
import org.zkdemo.messages.MessageType;

import akka.actor.UntypedActor;

public class Core extends UntypedActor implements Watcher
{
    
    private static final transient Logger LOGGER = Logger.getLogger(Core.class);
    
    private ApplicationContext            applicationContext;
    private ZooKeeper                     zooKeeper;
    
    private int                           sessionTimeout;
    private int                           port;
    private String                        host;
    
    public Core(ApplicationContext applicationContext)
    {
        this.applicationContext = applicationContext;
    }
    
    public Core(ApplicationContext applicationContext, String host, int port, int sessionTimeout)
    {
        this(applicationContext);
        this.sessionTimeout = sessionTimeout;
        this.port = port;
        this.host = host;
    }
    
    @Override
    public void preStart() throws Exception
    {
        postStop();
        zooKeeper = new ZooKeeper(host + ":" + port, sessionTimeout, this);
    }
    
    @Override
    public void postStop() throws Exception
    {
        if (zooKeeper != null)
        {
            zooKeeper.close();
        }
    }
    
    @Override
    public void onReceive(Object message) throws Exception
    {
        if (message instanceof Message)
        {
            Message msg = (Message) message;
            
            switch (msg.getType())
            {
                case MSG_CORE_STARTED:
                    break;
            }
        }
        
        unhandled(message);
    }
    
    @Override
    public void process(WatchedEvent event)
    {
        if (KeeperState.SyncConnected.equals(event.getState()))
        {
            getSelf().tell(new Message(MessageType.MSG_CORE_STARTED, null), getSelf());
        }
    }
}
